package com.springboot.kafka.producer;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class DemoInterceptor implements ProducerInterceptor <String,RecordBean>{

    private volatile long success = 0;
    private volatile long fail = 0;


    @Override
    public ProducerRecord<String, RecordBean> onSend(ProducerRecord<String, RecordBean> record) {
        RecordBean value = record.value();
        RecordBean recordBean = new RecordBean(value.getName() + "被拦截后的消息！", value.getAge());

        return new ProducerRecord<String, RecordBean>(record.topic(), record.partition(), record.timestamp()
                , record.key(), recordBean, record.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            success++;
        }
        if (metadata == null) {
            fail++;
        }
    }

    @Override
    public void close() {
        double v = (double) success / (success + fail);
        System.out.println("发送消息成功率为：" + v * 100 + "%");
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
